package net.maergaiyun.dwm;

import lombok.extern.slf4j.Slf4j;
import net.maergaiyun.func.AsyncLocationRequestFunction;
import net.maergaiyun.func.DeviceMapFunction;
import net.maergaiyun.func.LocationMapFunction;
import net.maergaiyun.model.ShortLinkWideDO;
import net.maergaiyun.util.KafkaUtil;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.concurrent.TimeUnit;

@Slf4j
public class DwmShortLinkWideApp {
    /**
     * 定义source topic
     */
    private static final String SOURCE_TOPIC = "dwd_link_visit_topic";

    /**
     * 定义sink topic
     */
    private static final String SINK_TOPIC = "dwm_link_visit_topic";

    /**
     * 定义消费者组
     */
    private static final String GROUP_ID = "dwm_short_link_group";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取流
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(SOURCE_TOPIC, GROUP_ID);
        DataStreamSource<String> ds = env.addSource(kafkaConsumer);

        // 格式转换, 补齐设备信息
        SingleOutputStreamOperator<ShortLinkWideDO> deviceWideDS = ds.map(new DeviceMapFunction());

        deviceWideDS.print("设备信息宽表补齐");

        // 补齐地理位置信息
//        SingleOutputStreamOperator<String> shortLinkWideDS = deviceWideDS.map(new LocationMapFunction());
        SingleOutputStreamOperator<String> shortLinkWideDS = AsyncDataStream.unorderedWait(deviceWideDS, new AsyncLocationRequestFunction(), 1000, TimeUnit.MILLISECONDS, 100);
        shortLinkWideDS.print("地理位置信息补齐");

        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);
        // 将sink写入dwm层
        shortLinkWideDS.addSink(kafkaProducer);

        env.execute();
    }

}
